// Copyright 2019 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// ------------------------------------------------------------------

#ifndef SRC_CPP_RTPS_TRANSPORT_TCPTRANSPORTINTERFACE_H_
#define SRC_CPP_RTPS_TRANSPORT_TCPTRANSPORTINTERFACE_H_

#include <transport/TransportInterface.h>
#include <transport/TCPTransportDescriptor.h>
#include <deps/common/IPFinder.h>
#include <tcp/RTCPHeader.h>
#include <TCPChannelResourceBasic.h>
#include <TCPAcceptorBasic.h>

#include <thread>
#include <vector>
#include <map>
#include <memory>
#include <mutex>

#include <asio.hpp>
#include <asio/steady_timer.hpp>

namespace vbs {
namespace transport {

class TCPChannelResource;

/**
 * This is a default TCP Interface implementation.
 *    - Opening an output channel by passing a remote locator will try to open a TCP conection with
 * the endpoint. If there is created a connection with the same endpoint, the transport will use the
 * same one.
 *
 *    - It is possible to provide a white list at construction, which limits the interfaces the
 * transport will ever be able to interact with. If left empty, all interfaces are allowed.
 *
 *    - Opening an input channel by passing a locator will open a socket listening on the given
 * physical port on every whitelisted interface, it will wait for incoming connections until the
 * receiver closes the channel. Several endpoints can connect to other to the same physical port,
 * because the OS creates a connection socket after each establishment.
 * @ingroup TRANSPORT_MODULE
 */
class TCPTransportInterface : public TransportInterface {
    class ReceiverInUseCV {
     public:
        bool in_use = false;

        std::condition_variable cv;
    };

    std::atomic<bool> alive_;

 protected:
    std::vector<vbsutil::xmlparser::IPFinder::info_IP> current_interfaces_;
    asio::io_service io_service_;
    // asio::io_service io_service_timers_;
    std::shared_ptr<std::thread> io_service_thread_;
    // std::shared_ptr<std::thread> io_service_timers_thread_;
    mutable std::mutex sockets_map_mutex_;
    mutable std::mutex receiver_resources_mutex_;
    mutable std::mutex send_resources_mutex_;
    mutable std::mutex acceptors_mutex_;

    std::map<Locator, std::shared_ptr<TCPChannelResource>> channel_resources_;  // The key is the "Physical locator"
    // The key is the physical port
    std::map<uint16_t, std::shared_ptr<ReceiverResource>> receiver_resources_;

    std::vector<std::pair<TCPChannelResource*, uint64_t>> sockets_timestamp_;

    // asio::steady_timer keep_alive_event_;

    std::map<Locator, std::shared_ptr<TCPAcceptor>> acceptors_;

    TCPTransportInterface(int32_t transport_kind);

    virtual bool compare_locator_ip(const Locator& lh, const Locator& rh) const = 0;

    virtual bool compare_locator_ip_and_port(const Locator& lh, const Locator& rh) const = 0;

    virtual void fill_local_ip(Locator& loc) const = 0;

    //! Closes the given p_channel_resource and unbind it from every resource.
    void close_tcp_socket(std::shared_ptr<TCPChannelResource>& channel);

    //! Creates a TCP acceptor to wait for incoming connections by the given locator.
    bool create_acceptor_socket(const Locator& locator);

    virtual void get_ips(std::vector<vbsutil::xmlparser::IPFinder::info_IP>& loc_names,
                         bool return_loopback = false) const = 0;

    bool is_input_port_open(uint16_t port) const;

    //! Functions to be called from new threads, which takes cares of performing a blocking receive
    void perform_listen_operation(std::weak_ptr<TCPChannelResource> channel);

    bool read_body(ertps::rtps::octet* receive_buffer, uint32_t receive_buffer_capacity, uint32_t* bytes_received,
                   std::shared_ptr<TCPChannelResource>& channel, std::size_t body_size);

    virtual void set_receive_buffer_size(uint32_t size) = 0;
    virtual void set_send_buffer_size(uint32_t size) = 0;

    void clean();  // Must be called on childs destructors!

    virtual void endpoint_to_locator(const asio::ip::tcp::endpoint& endpoint, Locator& locator) const = 0;

    /**
     * Applies TLS configuration to ssl_context
     *
     * @return true if everything worked fine, false otherwise.
     */
    bool apply_tls_config();

    /**
     * Aux method to retrieve cert password as a callback
     */
    std::string get_password() const;

    /**
     * Send a buffer to a destination
     */
    bool send(ertps::rtps::octet* send_buffer, uint32_t send_buffer_size, std::shared_ptr<TCPChannelResource>& channel,
              const Locator& remote_locator);

    bool async_send(octet* send_buffer, uint32_t send_buffer_size, std::shared_ptr<TCPChannelResource>& channel,
                    const Locator& remote_locator, AsyncSendHandler handler);

 public:
    virtual ~TCPTransportInterface();

    //! Stores the binding between the given locator and the given TCP socket. Server side.
    void bind_socket(std::shared_ptr<TCPChannelResource>&);

    //! Removes the listening socket for the specified port.
    bool CloseInputChannel(const Locator&) override;

    //! Removes all outbound sockets on the given port.
    void CloseOutputChannel(std::shared_ptr<TCPChannelResource>& channel);

    //! Reports whether Locators correspond to the same port.
    bool DoInputLocatorsMatch(const Locator&, const Locator&) const override;

    virtual asio::ip::tcp::endpoint generate_endpoint(uint16_t port) const = 0;

    virtual asio::ip::tcp::endpoint generate_endpoint(const Locator& loc, uint16_t port) const = 0;

    virtual asio::ip::tcp::endpoint generate_local_endpoint(Locator& loc, uint16_t port) const = 0;

    virtual asio::ip::tcp generate_protocol() const = 0;

    virtual asio::ip::tcp get_protocol_type() const = 0;

    bool init() override;

    //! Checks whether there are open and bound sockets for the given port.
    bool IsInputChannelOpen(const Locator&) const override;

    //! Checks for TCP kinds.
    bool IsLocatorSupported(const Locator&) const override;

    /** Opens an input channel to receive incoming connections.
     *   If there is an existing channel it registers the receiver resource.
     */
    bool OpenInputChannel(const Locator&, std::shared_ptr<ReceiverResource>, uint32_t) override;

    //! Opens a socket on the given address and port (as long as they are white listed).
    bool OpenOutputChannel(std::map<Locator_t, std::shared_ptr<SenderResource>>&, const Locator_t&,
                           Locator_t&) override;

    /**
     * Converts a given remote locator (that is, a locator referring to a remote
     * destination) to the main local locator whose channel can write to that
     * destination. In this case it will return a 0.0.0.0 address on that port.
     */
    Locator RemoteToMainLocal(const Locator&) const override;
    /**
     * Transforms a remote locator into a locator optimized for local communications.
     *
     * If the remote locator corresponds to one of the local interfaces, it is converted
     * to the corresponding local address.
     *
     * @param [in]  remote_locator Locator to be converted.
     * @param [out] result_locator Converted locator.
     *
     * @return false if the input locator is not supported/allowed by this transport, true
     * otherwise.
     */
    bool transform_remote_locator(const Locator& remote_locator, Locator& result_locator) const override;

    /**
     * Blocking Receive from the specified channel.
     * @param rtcp_manager pointer to the RTCP Manager.
     * @param channel pointer to the socket where the method is going to read the messages.
     * @param receive_buffer vector with enough capacity (not size) to accomodate a full receive
     * buffer. That capacity must not be less than the receive_buffer_size supplied to this class
     * during construction.
     * @param receive_buffer_capacity maximum size of the buffer.
     * @param[out] receive_buffer_size Size of the packet received.
     * @param[out] remote_locator associated remote locator.
     */
    bool Receive(std::shared_ptr<TCPChannelResource>& channel, ertps::rtps::octet* receive_buffer,
                 uint32_t receive_buffer_capacity, uint32_t& receive_buffer_size, Locator& remote_locator,
                 RTCPHeader& rtcp_header);

    /**
     * Blocking Send through the specified channel.
     * @param send_buffer Slice into the raw data to send.
     * @param send_buffer_size Size of the raw data. It will be used as a bounds check for the
     * previous argument. It must not exceed the send_buffer_size fed to this class during
     * construction.
     * @param channel channel we're sending from.
     * @param destination_locators_begin pointer to destination locators iterator begin, the
     * iterator can be advanced inside this fuction so should not be reuse.
     * @param destination_locators_end pointer to destination locators iterator end, the iterator
     * can be advanced inside this fuction so should not be reuse.
     */
    bool send(ertps::rtps::octet* send_buffer, uint32_t send_buffer_size, std::shared_ptr<TCPChannelResource>& channel,
              ertps::rtps::LocatorsIterator* destination_locators_begin,
              ertps::rtps::LocatorsIterator* destination_locators_end);

    bool async_send(octet* send_buffer, uint32_t send_buffer_size, std::shared_ptr<TCPChannelResource>& channel,
                    ertps::rtps::LocatorsIterator* destination_locators_begin,
                    ertps::rtps::LocatorsIterator* destination_locators_end, AsyncSendHandler handler);
    bool select_multicast() const override { return false; }

    //! Callback called each time that an incoming connection is accepted.
    void SocketAccepted(std::shared_ptr<asio::ip::tcp::socket> socket, const Locator& locator,
                        const asio::error_code& error);

    //! Callback called each time that an outgoing connection is established.
    void SocketConnected(const std::weak_ptr<TCPChannelResource>& channel, const asio::error_code& error);

    bool getDefaultMetatrafficMulticastLocators(LocatorList& locators,
                                                uint32_t metatraffic_multicast_port) const override;

    bool getDefaultMetatrafficUnicastLocators(LocatorList& locators, uint32_t metatraffic_unicast_port) const override;

    bool getDefaultUnicastLocators(LocatorList& locators, uint32_t unicast_port) const override;

    bool fillMetatrafficMulticastLocator(Locator& locator, uint32_t metatraffic_multicast_port) const override;

    bool fillMetatrafficUnicastLocator(Locator& locator, uint32_t metatraffic_unicast_port) const override;

    bool configureInitialPeerLocator(Locator& locator, const vbsutil::xmlparser::PortParameters& port_params,
                                     uint32_t domainId, LocatorList& list) const override;

    bool fillUnicastLocator(Locator& locator, uint32_t well_known_port) const override;
    uint32_t max_recv_buffer_size() const override { return configuration()->maxMessageSize; }

    // void DeleteSocket(TCPChannelResource* channelResource);

    virtual const TCPTransportDescriptor* configuration() const = 0;

    virtual TCPTransportDescriptor* configuration() = 0;

    void update_network_interfaces() override;
};

}  // namespace transport
}  // namespace vbs

#endif  // SRC_CPP_RTPS_TRANSPORT_TCPTRANSPORTINTERFACE_H_
